%%%-------------------------------------------------------------------
%%% @author wukai
%%% @copyright (C) 2020, <COMPANY>
%%% @doc
%%%
%%% @end
%%% Created : 29. 8月 2020 9:35 上午
%%%-------------------------------------------------------------------
-module(himq_mqtt_router).
-author("wukai").
-behaviour(gen_server2).
-include("himq.hrl").
-include_lib("stdlib/include/qlc.hrl").
-define(MAX_TOPIC_LEN,1024).
%% API
-export(
[init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3,handle_post_hibernate/1,handle_pre_hibernate/1]).
-define(ROUTER_TABLE, router_table).
%%路由表
-record(router_table, {filter :: string(), client :: string(), qos :: integer(), clean :: integer(), pid :: pid(),add_time::integer()}).
-export([init_router/1, start_sup/1, init_sup/2, add_router/5, del_router/3,dispatch/3,check_router/0]).
-export([match/2,subscribe_list/2]).
-record(state, {router :: atom()}).
init([]) ->
  {ok, #state{}, hibernate, {backoff, 1000, 1000, 10000}}.

handle_call(_Request, _From, State) ->
  {reply, ok, State}.

handle_cast(_Request, State) ->
  hibernate(State).

handle_info({dispatch,{From,Publish}}, State) ->
  delivery(From,Publish),
  hibernate(State);

handle_info(_Info, State) ->
  hibernate(State).

terminate(_Reason, _State) -> ok.

code_change(_OldVsn, State, _Extra) ->
  {ok, State}.

handle_post_hibernate(State)->
  %%lager:info("handle_post_hibernate"),
  hibernate(State).

handle_pre_hibernate(State)->
  %%lager:info("handle_pre_hibernate"),
  {hibernate, State}.

hibernate(State) ->
  {noreply, State, hibernate}.

dispatch(RouterPid,From,Publish)->RouterPid !{dispatch,{From,Publish}}.
delivery(From,{_, _, _, Topic, _}=Publish)->
    List =ets:lookup(?ROUTER_TABLE,Topic),
    case List of
      [] -> lager:info("no subscriber ~p", [Topic]);
      [_H|_T] ->
        lists:foreach(
        fun(E) ->
          send(From, E#router_table.client, Publish, E#router_table.pid, E#router_table.qos,E#router_table.clean)
        end,
        List)
    end.

send(From, To, Publish, Pid, NewQos,Clean) ->
  {Dup, _, Retain, Topic, Packet} = Publish,
  NewP = {Dup, NewQos, Retain, Topic, Packet},
  %%检测是否是自己订阅了离线或者上线消息
  case drop(Topic,Packet,list_to_binary(To)) of
    true->ok;
    %%clean=0 不保存消息 不保存会话
    false-> case {is_process_alive(Pid), Clean,Retain} of
              {false, 0,0} ->
                save(To, {From, NewP});
              {true, _,_} ->
                lager:info("publish ~p from ~p to ~p",[Publish,From,To]),
                Pid ! {pub,{From, NewP}};
              A ->
                lager:info("send drop ~p", [A])
            end
  end.

save(To, {From, {Dup, Qos, Retain, Topic, Packet}})->
  List_Node = atom_to_list(node()),
  List = lists:concat(List_Node++erlang:tuple_to_list(erlang:timestamp())++erlang:pid_to_list(self())),
  lager:info("List ~p",[List]),
  Uuid = himq:md5(List),
  Item = #himq_undelivered{
    id=To,
    uuid = Uuid,
    dup = Dup,
    qos = Qos,
    remark = <<"client offline">>,
    from = From,
    retain = Retain,
    step = dispatch,
    topic = Topic,
    packet = Packet,
    created_at = himq:timestamp()
  },
  case himq_mnesia:write(Item) of
    {atomic,ok}->ok;
    Err->lager:info("save undelivred msg error ~p",[Err])
  end.

%%删除发送给自己的自己掉线的消息
drop(<<"$sys/client/disconnect">>,Packet,To)->
  #{<<"id">>:=Who}=jsx:decode(Packet,[return_maps]),
  Who == To;
%%删除发送给自己的上线消息
drop(<<"$sys/client/connect">>,Packet,To)->
  #{<<"id">>:=Who}=jsx:decode(Packet,[return_maps]), Who == To;
drop(_,_,_)->false.


%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%初始化router sup%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
init_router(Num) ->
  RouterSup = #{id=>himq_mqtt_router_sup, start=>{?MODULE, start_sup, [Num]},
    type=>supervisor,
    restart=>permanent,
    shutdown=>brutal_kill
  },
  supervisor:start_child(himq_sup, RouterSup).

start_sup(Num) ->
  proc_lib:start_link(?MODULE, init_sup, [self(), Num]).

init_sup(Parent, Number) ->
  init_router_table(),
  process_flag(trap_exit, true),
  register(himq_mqtt_router_sup, self()),
  %%生成router进程
  List = gen_router(Number, []),
  {_,T1,_} = erlang:timestamp(),
  himq_eenv:set(mqtt_router, routers, List),
  {_,T2,_} = erlang:timestamp(),
  io:format("time: ~p", [T2-T1]),
  ok = proc_lib:init_ack(Parent, {ok, self()}),
  loop(self()).

init_router_table() ->
  case ets:info(?ROUTER_TABLE, size) of
    undefined ->
      ets:new(?ROUTER_TABLE, [bag, public, named_table,
        {write_concurrency, true}, {read_concurrency, true},
        {keypos, #router_table.filter}]);
    S when is_integer(S) -> ok
  end.

gen_router(N, List) when N > 0 ->
  {ok, P} = gen_server2:start_link(?MODULE, [], []),
  Pid = pid_to_list(P),
  gen_router(N - 1, List ++ [Pid]);
gen_router(0, List) -> List.
%%%%loop消息循环%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
loop(SupPid) ->
  receive
    {'EXIT', _Pid, _Reason} ->
      {ok, Pid} = gen_server2:start_link(himq_mqtt_router, [], []),
      io:format("new router Pid ~p", [Pid]),
      loop(SupPid);
    {add, FilterList,PacketId, ClientId, Pid, Clean} ->
      sub_topics(FilterList,PacketId,ClientId,Pid,Clean),
      loop(SupPid);
    {delete, FilterList, ClientId,Pid} ->
      unsub_topics(FilterList,ClientId,Pid),
      loop(SupPid);
    Other ->
      lager:warning("sup msg ~p \n", [Other]),
      loop(SupPid)
  end.

check_router() ->
  List = himq_eenv:get(mqtt_router, routers, []),
  Range = erlang:length(List),
  Index = rand:uniform(Range),
  Pid = list_to_pid(lists:nth(Index, List)),
  case is_process_alive(Pid) of
    true -> Pid;
    false -> check_router()
  end.

sub_topics(FilterList,PacketID, Client, Pid,Clean) ->
  List = [#router_table{client = Client, pid = Pid, qos = Qos,clean = Clean, filter = Filter,add_time = himq:timestamp()} || {Filter, Qos} <- FilterList],
  %%删除该client已经存在的订阅
  lists:foreach(fun(E) -> delete_old(E) end, List),
  Bin = check_write(Client, Pid, List, <<>>),
  Pid ! {sub_ack, PacketID, Bin,FilterList}.

unsub_topics(FilterList, Client, Pid) ->
  List = [#router_table{client = Client, pid = Pid, qos = '_', clean = '_', filter = Filter} || Filter <- FilterList],
  lists:foreach(fun(E) -> ets:match_delete(?ROUTER_TABLE, E) end, List).

delete_old(E) ->
  ets:match_delete(?ROUTER_TABLE, #router_table{client =
  E#router_table.client, filter = E#router_table.filter, _ = '_', _ = '_', _ = '_',_='_'}).

check_write(Client, Pid, [H|Tail], Bin) ->
  Qos = H#router_table.qos,
  V1 = validate({filter, H#router_table.filter}),
  V2 = qos_check(H#router_table.qos),
  V3 = wildcard( H#router_table.filter),
  Ret = case {V1, V2, V3} of
          {true, true, false} ->
            try
              ets:insert(?ROUTER_TABLE, H),
                %%发送retain消息
              %%okmq_mqtt_publish:retain({get,H#router_table.filter,Pid})
              <<Bin/binary,  Qos:8>>
            catch
              _:_ -> <<Bin/binary, 128:8>>
            end;
          _ -> <<Bin/binary, 128:8>>
        end,
  check_write(Client, Pid, Tail, Ret);

check_write(_, _, [], Bin) -> Bin.
qos_check(0) -> true;
qos_check(1) -> true;
qos_check(2) -> true;
qos_check(_) -> false.

add_router(FilterList, PacketId,Client, Pid, CleanFlag) ->
  himq_mqtt_router_sup ! {add, FilterList, PacketId,Client, Pid, CleanFlag}.
del_router(FilterList, ClientId,Pid) -> himq_mqtt_router_sup ! {delete, FilterList, ClientId,Pid}.

wildcard(Topic) when is_binary(Topic) ->
  wildcard(words(Topic));
wildcard([]) ->
  false;
wildcard(['#'|_]) ->
  true;
wildcard(['+'|_]) ->
  true;
wildcard([_H|T]) ->
  wildcard(T).

match(<<$$, _/binary>>, <<$+, _/binary>>) ->
  false;
match(<<$$, _/binary>>, <<$#, _/binary>>) ->
  false;
match(Name, Filter) when is_binary(Name) and is_binary(Filter) ->
  match(words(Name), words(Filter));
match([], []) ->
  true;
match([H|T1], [H|T2]) ->
  match(T1, T2);
match([_H|T1], ['+'|T2]) ->
  match(T1, T2);
match(_, ['#']) ->
  true;
match([_H1|_], [_H2|_]) ->
  false;
match([_H1|_], []) ->
  false;
match([], [_H|_T2]) ->
  false.

%%校验主题名称或者主题过滤器
validate({_, <<>>}) ->
  false;
validate({_, Topic}) when is_binary(Topic) and (size(Topic) > ?MAX_TOPIC_LEN) ->
  false;
validate({filter, Topic}) when is_binary(Topic) ->
  validate2(words(Topic));

validate({name, Topic}) when is_binary(Topic) ->
  Words = words(Topic),
  validate2(Words) and (not wildcard(Words)).

validate2([]) ->
  true;
validate2(['#']) -> % end with '#'
  true;
validate2(['#'|Words]) when length(Words) > 0 ->
  false;
validate2([''|Words]) ->
  validate2(Words);
validate2(['+'|Words]) ->
  validate2(Words);
validate2([W|Words]) ->
  case validate3(W) of true -> validate2(Words); false -> false end.

validate3(<<>>) ->
  true;
validate3(<<C/utf8, _Rest/binary>>) when C == $#; C == $+; C == 0 ->
  false;
validate3(<<_/utf8, Rest/binary>>) ->
  validate3(Rest).

words(Topic) when is_binary(Topic) ->
  [word(W) || W <- binary:split(Topic, <<"/">>, [global])].

word(<<>>)    -> '';
word(<<"+">>) -> '+';
word(<<"#">>) -> '#';
word(Bin)     -> Bin.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%router_sup%%%%%%%%%%%%%%%%%%%%%%%%%%%


subscribe_list(Page, PageSize) ->
  QH = qlc:q([Subscribe || Subscribe <- ets:table(?ROUTER_TABLE)]),
  Qc = qlc:cursor(QH),
  case Page of
    1 -> skip;

    _ -> qlc:next_answers(Qc, (Page - 1) * PageSize)
  end,
  Size = ets:info(?ROUTER_TABLE, size),
  case qlc:next_answers(Qc, PageSize) of
    [] ->
      #{<<"page">>=>Page,
        <<"count">>=>Size,
        <<"items">>=>[]
      };
    L ->
      lager:info("subscriber list ~p ",[L]),
      R = lists:map(fun(E) ->
      #{
        <<"client">>=>list_to_binary(E#router_table.client),
        <<"filter">>=>E#router_table.filter,
        <<"pid">>=>list_to_binary(pid_to_list(E#router_table.pid)),
        <<"clean">>=>E#router_table.clean,
        <<"created_at">>=>E#router_table.add_time,
        <<"qos">>=>E#router_table.qos
      }
                       end, L),

      #{<<"page">>=>Page,
        <<"count">>=>Size,
        <<"items">>=>R
      }
  end.
